// Project name:JavaRedis
// Creator:muhong
// Date time:2022/12/17,10:58 上午
// Name:SingleThreadEventLoop

package org.gmh.channel;

import org.gmh.util.concurrent.EventExecutorGroup;
import org.gmh.util.concurrent.RejectedExceptionHandler;
import org.gmh.util.concurrent.RejectedExceptionHandlers;
import org.gmh.util.concurrent.SingleThreadEventExecutor;

import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadFactory;

/**
 * 只有单个线程的loop处理器，继承来executor和eventloop
 */
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {

    protected static final int DEFAULT_MAX_PENDING_TASKS = 16;

    private final Queue<Runnable> tailTasks;

    protected SingleThreadEventLoop(EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {
        this(parent, threadFactory, addTaskWakesUp, DEFAULT_MAX_PENDING_TASKS, RejectedExceptionHandlers.reject());
    }

    protected SingleThreadEventLoop(EventExecutorGroup parent, ThreadFactory threadFactory,
                                    boolean addTaskWakesUp, int maxPendingTasks,
                                    RejectedExceptionHandler rejectedExceptionHandler) {
        super(parent, threadFactory, addTaskWakesUp, maxPendingTasks, rejectedExceptionHandler);
        tailTasks = newTaskQueue(maxPendingTasks);
    }

    protected SingleThreadEventLoop(EventExecutorGroup parent, Executor executor,
                                    boolean addTaskWakesUp, int maxPendingTasks,
                                    RejectedExceptionHandler rejectedExceptionHandler) {
        super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExceptionHandler);
        tailTasks = newTaskQueue(maxPendingTasks);
    }

    @Override
    public EventLoopGroup parent() {
        return (EventLoopGroup) super.parent();
    }

    @Override
    public EventLoop next() {
        return (EventLoop) super.next();
    }

    @Override
    public ChannelFuture register(Channel channel) {
        return register(new DefaultChannelPromise(channel, this));
    }

    @Override
    public ChannelFuture register(ChannelPromise promise) {
        promise.channel().unsafe().register(this, promise);
        return promise;
    }

    @Override
    protected void afterRunningAllTasks() {
        runAllTaskFrom(tailTasks);
    }

    @Override
    protected boolean hasTasks() {
        return super.hasTasks() || !tailTasks.isEmpty();
    }

    @Override
    public int pendingTasks() {
        return super.pendingTasks() + tailTasks.size();
    }

    interface NonWakeupRunnable extends Runnable {

    }
}
